/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package aiaudio.processing.ratings.normilize;

import aiaudio.processing.base.mapreduce.CoordinateDoubleValueKey;
import aiaudio.processing.base.mapreduce.standard.mappers.UserToArtistValueMatrixMapper;
import aiaudio.processing.base.mapreduce.standard.mappers.UserToArtistValueRatingTableMapper;
import aiaudio.processing.MainTableGroup;
import aiaudio.processing.ratings.normilize.defaults.reducers.SimpleRatingUserToArtistValueReducerDefaultAvgOverUser;
import aiaudio.processing.ratings.normilize.defaults.reducers.SimpleRatingUserToArtistValueReducerDefaultZero;
import aiaudio.processing.ratings.normilize.zscore.reducers.ZScoreUserToArtistValueReducerDefaultAvgOverUser;
import aiaudio.processing.ratings.normilize.zscore.reducers.ZScoreUserToArtistValueReducerDefaultZero;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.mapreduce.TableReducer;

/**
 *
 * @author nastya
 */
public class RatingAlgorithmDefinition {

    protected String inputTableName;
    protected Class<? extends TableMapper<ImmutableBytesWritable, CoordinateDoubleValueKey>> mapperClass;
    protected Class<? extends TableReducer<ImmutableBytesWritable, CoordinateDoubleValueKey, ImmutableBytesWritable>> reducerClass;

    public RatingAlgorithmDefinition(String inputTableName, Class<? extends TableMapper<ImmutableBytesWritable, CoordinateDoubleValueKey>> mapperClass, Class<? extends TableReducer<ImmutableBytesWritable, CoordinateDoubleValueKey, ImmutableBytesWritable>> reducerClass) {
        this.inputTableName = inputTableName;
        this.mapperClass = mapperClass;
        this.reducerClass = reducerClass;
    }

    private RatingAlgorithmDefinition() {
    }

    public String getInputTableName() {
        return inputTableName;
    }

    public Class<? extends TableMapper<ImmutableBytesWritable, CoordinateDoubleValueKey>> getMapperClass() {
        return mapperClass;
    }

    public Class<? extends TableReducer<ImmutableBytesWritable, CoordinateDoubleValueKey, ImmutableBytesWritable>> getReducerClass() {
        return reducerClass;
    }

    public static RatingAlgorithmDefinition createDefinition(InitialData initialData, Normalizaton normalizaton, DefaultValue defaultValue, MainTableGroup tableGroup) {
        RatingAlgorithmDefinition def = new RatingAlgorithmDefinition();

        switch (initialData) {
            case INVERSED_LISTEN_COUNT:
                def.inputTableName = tableGroup.getPrepareRatingsAlgTableGroup().getInversedRatingMatrix();
                def.mapperClass = UserToArtistValueMatrixMapper.class;
                break;
            case LISTEN_COUNT:
            default:
                def.inputTableName = tableGroup.getInitialDataTableGroup().getRatingListTable();
                def.mapperClass = UserToArtistValueRatingTableMapper.class;
                break;
        }


        Class<? extends TableReducer<ImmutableBytesWritable, CoordinateDoubleValueKey, ImmutableBytesWritable>> reducerClass 
                = getReducer(normalizaton, defaultValue);

        def.reducerClass = reducerClass;

        return def;
    }
    
    private static Class<? extends TableReducer<ImmutableBytesWritable, CoordinateDoubleValueKey, ImmutableBytesWritable>> getReducer(Normalizaton normalizaton, DefaultValue defaultValue) {
        Class<? extends TableReducer<ImmutableBytesWritable, CoordinateDoubleValueKey, ImmutableBytesWritable>> reducerClass;
        reducerClass = ZScoreUserToArtistValueReducerDefaultAvgOverUser.class;
        if (normalizaton == Normalizaton.SIMPLE && defaultValue == DefaultValue.ZERO) {
            reducerClass = SimpleRatingUserToArtistValueReducerDefaultZero.class;
        }
        if (normalizaton == Normalizaton.SIMPLE && defaultValue == DefaultValue.AVG_OVER_USER) {
            reducerClass = SimpleRatingUserToArtistValueReducerDefaultAvgOverUser.class;
        }
        if (normalizaton == Normalizaton.ZSCORE && defaultValue == DefaultValue.ZERO) {
            reducerClass = ZScoreUserToArtistValueReducerDefaultZero.class;
        }
        if (normalizaton == Normalizaton.ZSCORE && defaultValue == DefaultValue.AVG_OVER_USER) {
            reducerClass = ZScoreUserToArtistValueReducerDefaultAvgOverUser.class;
        }
        return reducerClass;
    }
    
}
